package com.yunhe.abnormal.websocket;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.model.msg.AlarmEventMsgDTO;
import com.yunhe.common.model.data.AlarmEventDTO;
import com.yunhe.abnormal.service.data.AlarmEventService;
import com.yunhe.common.util.JsonUtil;
import com.yunhe.common.util.StringUtil;
import com.yunhe.common.websocket.WebsocketRealtimeHandler;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.util.*;

/**
 * 异常按等级统计WebSocket处理类
 * @author liuronglei
 */
public class AlarmCountHandler extends WebsocketRealtimeHandler {
    private static final Logger logger = LoggerFactory.getLogger(AlarmCountHandler.class);
    private ObjectMapper mapper = JsonUtil.getObjectMapper();
    @Autowired
    private AlarmEventService alarmEventService;

    @Override
    public Long refreshInterval() {
        return 2000L;
    }

    @Override
    public void beforeSend(WebSocketSession session, Object content) {

    }

    @Override
    public String getPointField() {
        return "stationIds";
    }

    @Override
    public void afterEstablished(WebSocketSession session) {

    }

    @Override
    public void afterClosed(WebSocketSession session, CloseStatus status) {

    }

    @Override
    public void handleMessage(WebSocketSession session, TextMessage message, List<String> points) {
        Map<String, Map<String, Integer>> result = new HashMap<>();
        for (String point : points) {
            Map<String, Integer> onResult = alarmEventService.realTimeWarningCountByStationIds(Collections.singletonList(Long.parseLong(point)));
            result.put(point, onResult);
        }
        sendMessage(result, session, points);
    }

    /**
     * 监听告警内容刷新消息
     * @param record 消息内容：主要是电站ID
     */
    @KafkaListener(topics = KafkaConsts.Topic.AlarmEventRefresh, groupId = KafkaConsts.Group.RandomGroupId)
    public void alarmCountListener(ConsumerRecord<?, String> record) {
        try {
            String value = record.value();
            if (!StringUtil.isEmpty(value)) {
                AlarmEventMsgDTO msg = mapper.readValue(value, AlarmEventMsgDTO.class);
                List<AlarmEventDTO> alarmEvents = msg.getAlarmEvents();
                Map<String, Map<String, Integer>> result = new HashMap<>();
                // 判断是否需要刷新全部电站
                boolean refreshAll = false;
                for (AlarmEventDTO alarmEvent : alarmEvents) {
                    // 只要有一个消息没有电站ID，刷新全部电站
                    Long stationId = alarmEvent.getStationId();
                    if (stationId == null) {
                        refreshAll = true;
                        break;
                    }
                }
                List<String> points;
                if (refreshAll) {
                    // 取全部电站ID
                    points = getAllPoints();
                    for (String point : points) {
                        Long stationId = Long.parseLong(point);
                        Map<String, Integer> onResult = alarmEventService.realTimeWarningCountByStationIds(Collections.singletonList(stationId));
                        result.put(point, onResult);
                    }
                } else {
                    points = new ArrayList<>();
                    for (AlarmEventDTO alarmEvent : alarmEvents) {
                        Long stationId = alarmEvent.getStationId();
                        if (stationId != null) {
                            String point = String.valueOf(stationId);
                            if (!points.contains(point)) {
                                points.add(point);
                                Map<String, Integer> onResult = alarmEventService.realTimeWarningCountByStationIds(Collections.singletonList(stationId));
                                result.put(point, onResult);
                            }
                        }
                    }
                }

                sendMessage(result, null, points);
            }
        } catch (Exception e) {
            logger.error("刷新异常按等级统计报错，{}，{}", record.value(), e);
        }
    }
}